package com.ass.rocketmq.producer;


import com.alibaba.fastjson.JSON;
import com.ass.rocketmq.base.AbstractMessageQueueSelector;
import com.ass.rocketmq.constant.MQTopic;
import com.ass.rocketmq.utils.MessageUtil;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationPreparedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;


@Log4j2
@Component
public abstract class AbstractDefaultProducer<M> {

    /**为了设置生产组 想要设置生产组下所有的生产者都是同一类的
    为了实现 “如果发送的是事物消息且原始生产者在发送之后崩溃，
     则Broker服务器会联系同一生产者组的其他生产者实例以提交或回溯消费” */


    @Value("${spring.profiles.active}")
    private String active;

    @Value("${rocketmq.accessKey}")
    private String accessKey;

    @Value("${rocketmq.secretKey}")
    private String secretKey;

    @Value("${rocketmq.namespace}")
    private String namespace;


    @Value("${rocketmq.name-server}")
    private String namesrvAddr;

    private DefaultMQProducer defaultMQProducer;

    protected abstract MQTopic getMqTopic();

    //因为每个子类都要启动一个对应的生产者 所以要给一个实例名称
    protected abstract String getInstanceName();


    /**
     * 发送消息 直接配置参数在枚举里面，对应各个属性，不要写多个send方法
     */
    public void send(String keys,M m) {
        try {
            Message message= MessageUtil.buildMessage(keys,JSON.toJSONString(m),getMqTopic(),active);
            SendResult sendResult = defaultMQProducer.send(message);
            log.info("sendStatus：{}",sendResult.getSendStatus());
        } catch (Exception e) {
            log.error(e);
        }
    }




    /**顺序消息 业务场景 在有序事件处理、撮合交易、数据实时增量同步等场景下
    例如创建订单的场景，需要保证同一个订单的生成、付款和发货，这三个操作被顺序执行。
     如果是普通消息，订单A的消息可能会被轮询发送到不同的队列中，不同队列的消息将无法保持顺序，
     而顺序消息发送时将ShardingKey相同（同一订单号）的消息序路由到一个逻辑队列中。相同的业务且保证顺序必须得发到同一个队列当中*/
    /**
     * @describe：顺序消息
     * @author：李勇
     * @date：2024/2/5
     * @param keys 做幂等的键
     * @param m 业务实体
     * @param selector 继承AbstractMessageQueueSelector的子类 实现顺序发送的逻辑
     * @param arg 具体参数配合要往哪个队列发送消息
     * @return：void
     * <Selector extends AbstractMessageQueueSelector> ：作用于这个方法，可以给这个方法入这个参数Selector
     * 并且必须是继承AbstractMessageQueueSelector
     */
    public <Selector extends AbstractMessageQueueSelector> void send(String keys, M m, Selector selector,Object arg) {
        try {
            MQTopic mqTopic = getMqTopic();
            Message message=MessageUtil.buildMessage(keys,JSON.toJSONString(m),mqTopic,active);
            SendResult sendResult = defaultMQProducer.send(message,selector,arg);
            log.info("sendStatus：{}",sendResult.getSendStatus());
        } catch (Exception e) {
            log.error(e);
        }
    }




    //批量发送
    public void send(String keys, List<M> list){
        MQTopic mqTopic = getMqTopic();
        List<Message> messages = list.stream()
                .filter(Objects::nonNull)
                .map(map -> MessageUtil.buildMessage(keys,JSON.toJSONString(map),mqTopic,active))
                .collect(Collectors.toList());
        try {
            SendResult sendResult =  defaultMQProducer.send(messages);
            log.info("sendStatus：{}",sendResult.getSendStatus());
        } catch (Exception e) {
            log.error(e);
        }
    }


    /**
     * 初始化mq生产者
     */
    private void initProducer() throws MQClientException {
        log.info("defaultProducer 正在创建---------------------------------------");
        MQTopic mqTopic = getMqTopic();
        String prefix = active + "_";
        defaultMQProducer = new DefaultMQProducer(namespace,
                prefix + mqTopic.tag,
                new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)));
        defaultMQProducer.setNamesrvAddr(namesrvAddr);
        defaultMQProducer.setVipChannelEnabled(false);
        defaultMQProducer.setRetryTimesWhenSendAsyncFailed(10);
        defaultMQProducer.setInstanceName(getInstanceName());
        defaultMQProducer.start();
        log.info("defaultProducer 启动成功---------------------------------------");
    }

    /**
     * 启动生产者 要在spring 容器启动后 进行一些监听 以遍可以从ioc容器中 拿到想要的bean，
     * 否则使用@PostConstruct 注解 因为此注解是在bean初始化的时候执行 有一些业务的bean还没被加载到ioc里面
     * 可能有一些bean 加载不到
     *
     * @param event
     */
    @EventListener(ApplicationPreparedEvent.class)
    public void startProducer(ApplicationPreparedEvent event) {
        //获取bean
//        PushConsumerProperty property = event.getApplicationContext().getBean(PushConsumerProperty.class);
        try {
            initProducer();
        } catch (MQClientException e) {
            log.error("[生产者start异常], nameServerAddress={}", namesrvAddr);
//            System.exit(-1);
        }
    }


}
